package com.rds.benchmark;

import redis.clients.jedis.*;

import java.io.*;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.Random;

public class RdsBenchmarkSingle {
    private final static int DEFAULT_THREADS = 100;

    private final static int SUBCONTENTS = 1000;

    private volatile static long DURATION = 120 * 1000;

    private final static Random RANDOM = new Random();
    private volatile static int Start = 0;
    private volatile static int Stop = 16383;

    private static void Usage() {
        System.out.println("\nUsage: java -jar runner-1.x.x.jar [-h host] [-p port] [-a password] [-t thread] [-c cmd] [--time duration] [--range start-stop] [stable]\n");
        System.out.println(" -h or --host   <hostname>           Server hostname (default 127.0.0.1)");
        System.out.println(" -p or --port   <port>               Server port (default 6379)");
        System.out.println(" -a or --pass   <password>           The password to login (default null)");
        System.out.println(" -t or --thread   <clients>          Number of parallel connections (default " + DEFAULT_THREADS + ")");
        System.out.println(" --time   <duration>                 Test run time, in seconds (default 120)");
        System.out.println(" --range  <start-stop>               The slot range of the key (default 0-16383)");
        System.out.println(" -c or --command  <cmd> <cmd-list>   Data type to test, optional values: string, list, set, hash, zset. publish, custom \\");
        System.out.println("                                     , all(include string,list,set,hash,zset,publish), (default string)");
        System.out.println("                                     The custom option is used to specify the test command, for example: [-c custom 'del aaa,set aaa bbb,get aaa'] ");
        System.out.println("                                     -c is used during performance test");
        System.out.println(" stable                              7 * 24 stability test (default performance test)");
        System.out.println(" --help                              this message");
        System.out.println("");
        System.exit(0);
    }

    private static String createKey(int no) {
        String key = "thread-123456789012345678901234567890123456789012345678901234567890-" + no;
        int slot = CRC16.getSlot(key.getBytes(StandardCharsets.UTF_8));
        String old_key = key;
        while (slot < Start || slot > Stop) {
            key = old_key + RANDOM.nextInt(100000);
            slot = CRC16.getSlot(key.getBytes(StandardCharsets.UTF_8));
        }
        return key;
    }

    private static void pString(final String host, final int port, final int max_threads, final String pass) throws InterruptedException {

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String key = createKey(idx);
                        String value = "value-"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890";

                        long stop = System.currentTimeMillis() + DURATION;
                        while (stop > System.currentTimeMillis()) {
                            Jedis conn = pool.getResource();
                            try {
                                conn.set(key, value);
                                synchronized (lockers[idx]) {
                                    times[idx]++;
                                }
                                String msg = conn.get(key);
                                if (!value.equals(msg)) {
                                    throw new IOException("set value = '" + value + "', but get value = '" + msg + "'");
                                }
                                synchronized (lockers[idx]) {
                                    times[idx]++;
                                }
                                conn.del(key);
                                synchronized (lockers[idx]) {
                                    times[idx]++;
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            }
                            conn.close();
                        }
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        Thread state = new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance string-testing(set,get,del) start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            for (int i = 0; i < max_threads; i++) {
                try {
                    threads[i].interrupt();
                } catch (Throwable t) {
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        });

        state.start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }

        state.join();
    }

    private static void pList(final String host, final int port, final int max_threads, final String pass) throws InterruptedException {

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String key = createKey(idx);
                        String value = "value-"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890";
                        long stop = System.currentTimeMillis() + DURATION;
                        Jedis conn = pool.getResource();
                        int count = SUBCONTENTS >> 1;
                        conn.del(key);
                        conn.close();
                        while (stop > System.currentTimeMillis()) {
                            conn = pool.getResource();
                            try {
                                for (int i = 0; i < count; i++) {
                                    conn.lpush(key, value);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                    conn.rpush(key, value);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                                for (int i = 0; i < count; i++) {
                                    conn.lpop(key);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                    conn.rpop(key);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            } finally {
                                conn.close();
                            }
                        }

                        conn.close();
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        Thread state = new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance list-testing(lpush,rpush,lpop,rpop) start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            for (int i = 0; i < max_threads; i++) {
                try {
                    threads[i].interrupt();
                } catch (Throwable t) {
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        });
        state.start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }

        state.join();
    }

    private static void pSet(final String host, final int port, final int max_threads, final String pass) throws InterruptedException {

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String key = createKey(idx);
                        String[] mkey = new String[SUBCONTENTS];
                        for (int i = 0; i < SUBCONTENTS; i++) {
                            mkey[i] = "thread-123456789012345678901234567890123456789012345678901234567890-" + idx + "--" + i;
                        }
                        long stop = System.currentTimeMillis() + DURATION;
                        Jedis conn = pool.getResource();
                        conn.del(key);
                        conn.close();
                        while (stop > System.currentTimeMillis()) {
                            conn = pool.getResource();
                            try {
                                for (int i = 0; i < SUBCONTENTS; ++i) {
                                    conn.sadd(key, mkey[i]);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                                for (int i = 0; i < SUBCONTENTS; ++i) {
                                    conn.srem(key, mkey[i]);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            } finally {
                                conn.close();
                            }
                        }

                        conn.close();
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        Thread state = new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance set-testing(sadd,srem) start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            for (int i = 0; i < max_threads; i++) {
                try {
                    threads[i].interrupt();
                } catch (Throwable t) {
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        });
        state.start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }

        state.join();
    }

    private static void pHash(final String host, final int port, final int max_threads, final String pass) throws InterruptedException {

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String key = createKey(idx);
                        String[] mkey = new String[SUBCONTENTS];
                        String value = "value-"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890";
                        for (int i = 0; i < SUBCONTENTS; i++) {
                            mkey[i] = "thread-123456789012345678901234567890123456789012345678901234567890-" + idx + "--" + i;
                        }
                        long stop = System.currentTimeMillis() + DURATION;
                        Jedis conn = pool.getResource();
                        conn.del(key);
                        conn.close();
                        while (stop > System.currentTimeMillis()) {
                            conn = pool.getResource();
                            try {
                                for (int i = 0; i < SUBCONTENTS; ++i) {
                                    conn.hset(key, mkey[i], value);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                                for (int i = 0; i < SUBCONTENTS; ++i) {
                                    conn.hdel(key, mkey[i]);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            } finally {
                                conn.close();
                            }
                        }

                        conn.close();
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        Thread state = new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance hash-testing(hset,hdel) start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            for (int i = 0; i < max_threads; i++) {
                try {
                    threads[i].interrupt();
                } catch (Throwable t) {
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        });
        state.start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }

        state.join();
    }

    private static void pZset(final String host, final int port, final int max_threads, final String pass) throws InterruptedException {

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String key = createKey(idx);
                        String[] mkey = new String[SUBCONTENTS];
                        for (int i = 0; i < SUBCONTENTS; i++) {
                            mkey[i] = "thread-123456789012345678901234567890123456789012345678901234567890-" + idx + "--" + i;
                        }
                        long stop = System.currentTimeMillis() + DURATION;
                        Jedis conn = pool.getResource();
                        conn.del(key);
                        conn.close();
                        while (stop > System.currentTimeMillis()) {
                            conn = pool.getResource();
                            try {
                                for (int i = 0; i < SUBCONTENTS; ++i) {
                                    conn.zadd(key, i, mkey[i]);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                                for (int i = 0; i < SUBCONTENTS; ++i) {
                                    conn.zrem(key, mkey[i]);
                                    synchronized (lockers[idx]) {
                                        times[idx]++;
                                    }
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            } finally {
                                conn.close();
                            }
                        }

                        conn.close();
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        Thread state = new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance zset-testing(zadd,zrem) start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            for (int i = 0; i < max_threads; i++) {
                try {
                    threads[i].interrupt();
                } catch (Throwable t) {
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        });
        state.start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }

        state.join();
    }

    private static void pPublish(final String host, final int port, final int max_threads, final String pass) throws InterruptedException {

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String key = createKey(idx);
                        String value = "value-"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890"
                                + "123456789012345678901234567890123456789012345678901234567890";

                        long stop = System.currentTimeMillis() + DURATION;
                        while (stop > System.currentTimeMillis()) {
                            Jedis conn = pool.getResource();
                            try {
                                conn.publish(key, value);
                                synchronized (lockers[idx]) {
                                    times[idx]++;
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            }
                            conn.close();
                        }
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        Thread state = new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance publish-testing(publish) start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            for (int i = 0; i < max_threads; i++) {
                try {
                    threads[i].interrupt();
                } catch (Throwable t) {
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        });
        state.start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }

        state.join();
    }

    private static void readResponse(BufferedReader reader) throws IOException {
        String msg = reader.readLine();
        if (msg.charAt(0) == '*') {
            int n = Integer.parseInt(msg.substring(1));
            for (int i = 0; i < n; ++i) {
                readResponse(reader);
            }
        } else if (msg.charAt(0) == '-') {
            throw new IOException(msg.substring(1));
        }
    }

    private static void pCustom(final String host, final int port, final int max_threads, final String pass, ArrayList<String> cmds) throws IOException {

        if (cmds == null || cmds.size() == 0) {
            throw new NullPointerException("command is null");
        }

        System.out.println("Connect to '" + host + ":" + port + "'");

        Thread[] threads = new Thread[max_threads];
        final Object[] lockers = new Object[max_threads];
        final int[] times = new int[max_threads];

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(max_threads);
        poolConfig.setMaxIdle(max_threads);

        final JedisPool pool = new JedisPool(poolConfig, host, port, null, pass);

        Jedis jedis = pool.getResource();
        jedis.dbSize();
        jedis.close();
        pool.close();

        Random random = new Random();

        for (int i = 0; i < max_threads; i++) {
            lockers[i] = new Object();
            times[i] = 0;
            Socket socket = new Socket();
            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
            if (pass != null) {
                writer.write("*2\r\n$4\r\nauth\r\n$" + pass.length() + "\r\n" + pass + "\r\n");
                writer.flush();
                readResponse(reader);
            }

            threads[i] = new Thread(Integer.toString(i)) {
                @Override
                public void run() {
                    int idx = Integer.parseInt(Thread.currentThread().getName());
                    try {
                        String cmd = cmds.get(random.nextInt(cmds.size()));

                        long stop = System.currentTimeMillis() + DURATION;
                        while (stop > System.currentTimeMillis()) {
                            try {
                                writer.write(cmd);
                                writer.newLine();
                                writer.flush();
                                readResponse(reader);
                                synchronized (lockers[idx]) {
                                    times[idx]++;
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                break;
                            }
                        }

//                        conn.close();
                    } finally {
                        synchronized (lockers[idx]) {
                            times[idx] = -1;
                        }
                    }
                }
            };
        }

        new Thread(() -> {
            long count = 0;
            long size = 0;
            boolean notFirst = false;
            long first = 0;
            long second = 0;

            System.out.println("\nPerformance string-testing start...");
            boolean running = true;
            while (running) {
                long begin = System.currentTimeMillis();

                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                running = false;
                int cur = 0;
                for (int i = 0; i < max_threads; i++) {
                    synchronized (lockers[i]) {
                        int one = times[i];
                        if (one >= 0) {
                            cur += one;
                            times[i] = 0;
                            running = true;
                        }
                    }
                }
                if (running) {
                    long duration = (System.currentTimeMillis() - begin);
                    second = first;
                    first = (long) (cur * 1000.0 / duration);
                    System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");
                    if (second > 0) {
                        if (!notFirst) {
                            // 去掉第一条记录
                            notFirst = true;
                        } else {
                            // second 为了去掉最后1条记录
                            count += second;
                            ++size;
                        }
                    }
                }
            }
            System.out.println("Performance testing end, average " + (size > 0 ? count / size : 0) + "/s\n");
        }).start();

        for (int i = 0; i < max_threads; i++) {
            threads[i].start();
        }
    }

    private static void stable(final String host, final int port, final String pass) {
        String basekey = "Stb-123456789012345678901234567890123456789012345678901234567890";
        String value = "value-"
                + "123456789012345678901234567890123456789012345678901234567890"
                + "123456789012345678901234567890123456789012345678901234567890"
                + "123456789012345678901234567890123456789012345678901234567890"
                + "123456789012345678901234567890123456789012345678901234567890";
        Jedis conn = new Jedis(new HostAndPort(host, port), DefaultJedisClientConfig.builder().password(pass).build());

        conn.flushDB();
        System.out.println("FlushDB ok.");

        // 7*24
        long stop = System.currentTimeMillis() + 7 * 24 * 60 * 60 * 1000;

        Random random = new Random();
        System.out.println("Connect to '" + host + ":" + port);

        System.out.println("\nStable testing for 7 * 24 start...");
        long last_output_time = System.currentTimeMillis();
        while (stop >= System.currentTimeMillis()) {
            int loop = random.nextInt(1000) + 10;
            try {
                for (int i = 0; i < loop; ++i) {
                    int type = i % 12;
                    String key = basekey + i;
                    String tmpValue = value + i;
                    String v;
                    boolean ok;
                    double d;
                    switch (type) {
                        case 1: // List
                        case 2: // List
                            conn.lpush(key, tmpValue);
                            conn.rpush(key, value);
                            v = conn.lpop(key);
                            conn.rpop(key);
                            if (!tmpValue.equals(v)) {
                                throw new IOException("lpush value = '" + tmpValue + "', but lpop value = '" + v + "'");
                            }
                            break;
                        case 3: // set
                        case 4: // set
                        case 5: // set
                            conn.sadd(key, tmpValue);
                            conn.sadd(key, value);
                            ok = conn.sismember(key, tmpValue);
                            if ((type & 0xe0) == 0) {
                                // 被清空
                                conn.srem(key, tmpValue, value);
                            } else {
                                // 会留1项
                                conn.srem(key, tmpValue);
                            }
                            if (!ok) {
                                throw new IOException("sismember failed, value = '" + tmpValue + "'");
                            }
                            break;
                        case 6: // hash
                        case 7: // hash
                        case 8: // hash
                            conn.hset(key, tmpValue, tmpValue);
                            conn.hset(key, value, "value");
                            v = conn.hget(key, tmpValue);
                            if ((type & 0xe0) == 0) {
                                // 被清空
                                conn.hdel(key, value, tmpValue);
                            } else {
                                // 留1项
                                conn.hdel(key, value);
                            }
                            if (!tmpValue.equals(v)) {
                                throw new IOException("hget value = '" + tmpValue + "', but hset value = '" + v + "'");
                            }
                            break;
                        case 9:
                        case 10:
                            conn.zadd(key, 0.5, tmpValue);
                            conn.zadd(key, 0, value);
                            d = conn.zscore(key, tmpValue);
                            conn.zrem(key, tmpValue);
                            if (d > 0.500001 || d < 0.499999) {
                                throw new IOException("zset value = '" + 0.5 + "', but zscore value = '" + d + "'");
                            }
                            break;
                        case 11:
                            conn.publish("channel", "message");
                            conn.publish("channel", "message");
                            conn.publish("channel" + i, "message" + i);
                            conn.publish("channel" + i, "message" + i);
                        default:// Strng
                            conn.set(key, value);
                            conn.set(key, tmpValue);
                            v = conn.get(key);
                            conn.del(key);
                            if (!tmpValue.equals(v)) {
                                throw new IOException("set value = '" + tmpValue + "', but get value = '" + v + "'");
                            }
                    }
                }
                // 每分钟
                if ((System.currentTimeMillis() - last_output_time) > 60 * 1000) {
                    System.out.println(new Date() + " test ok.");
                    last_output_time = System.currentTimeMillis();
                }
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println(new Date() + " error: " + e + ", reconnecting...");
                try {
                    conn.close();
                } catch (Throwable t2) {
                }
                try {
                    conn = new Jedis(new HostAndPort(host, port), DefaultJedisClientConfig.builder().password(pass).build());
                } catch (Throwable t2) {
                }
            }
            try {
                Thread.sleep(loop);
            } catch (Exception e) {
                System.out.println(new Date() + " thread breaked: " + e);
                break;
            }
        }
        conn.close();
    }

    public static void main(String[] argv) {
        String host = "localhost";
        int port = -1;
        int branch = 1;
        int threads = DEFAULT_THREADS;
        int cmd = 0;
        ArrayList<String> custom_cmds = new ArrayList<>();

        String password = null;

        // 解析参数
        for (int i = 0; i < argv.length; i++) {
            if ("-h".equalsIgnoreCase(argv[i])
                    || "--host".equalsIgnoreCase(argv[i])) {
                if (i + 1 < argv.length) {
                    host = argv[++i];
                } else {
                    System.out.println("invalid parameter -h");
                    Usage();
                }
            } else if ("-p".equalsIgnoreCase(argv[i])
                    || "--port".equalsIgnoreCase(argv[i])) {
                try {
                    port = Integer.parseInt(argv[++i]);
                } catch (Exception e) {
                    System.out.println("invalid parameter -p: " + e.getMessage());
                    Usage();
                }
            } else if ("-t".equalsIgnoreCase(argv[i])
                    || "--thread".equalsIgnoreCase(argv[i])) {
                try {
                    threads = Integer.parseInt(argv[++i]);
                } catch (Exception e) {
                    System.out.println("invalid parameter -t: " + e.getMessage());
                    Usage();
                }
            } else if ("-c".equalsIgnoreCase(argv[i])
                    || "--command".equalsIgnoreCase(argv[i])) {
                String command = argv[++i];
                if ("list".equalsIgnoreCase(command)) {
                    cmd = 1;
                } else if ("set".equalsIgnoreCase(command)) {
                    cmd = 2;
                } else if ("hash".equalsIgnoreCase(command)) {
                    cmd = 3;
                } else if ("zset".equalsIgnoreCase(command)) {
                    cmd = 4;
                } else if ("publish".equalsIgnoreCase(command)) {
                    cmd = 5;
                } else if ("custom".equalsIgnoreCase(command)) {
                    cmd = 10;
                    String[] cmds = argv[++i].split(",");
                    for (String s : cmds) {
                        s = s.trim();
                        if (s.length() > 0) {
                            custom_cmds.add(s);
                            System.out.println("Customized test command: '" + s + "' is loaded.");
                        }
                    }
                } else if ("all".equalsIgnoreCase(command)) {
                    cmd = 100;
                }
            } else if ("--time".equalsIgnoreCase(argv[i])) {
                try {
                    DURATION = Long.parseLong(argv[++i]) * 1000;
                } catch (Exception e) {
                    System.out.println("invalid parameter --time: " + e.getMessage());
                    Usage();
                }
            } else if ("--range".equalsIgnoreCase(argv[i])) {
                String range = argv[++i];
                try {
                    String[] rs = range.split("-");
                    if (rs != null && rs.length > 0) {
                        Start = Integer.parseInt(rs[0]);
                        if (rs.length > 1) {
                            Stop = Integer.parseInt(rs[1]);
                        }
                    }
                } catch (Throwable t) {
                    System.err.println("Error range: " + range);
                }
//                System.out.println("Range of slot is [" + Start + " - " + Stop + "]");
            } else if ("performance".equalsIgnoreCase(argv[i])) {
                branch = 1;
            } else if ("stable".equalsIgnoreCase(argv[i])) {
                branch = 2;
            } else if ("-a".equalsIgnoreCase(argv[i]) || "--pass".equalsIgnoreCase(argv[i])) {
                password = argv[++i];
            } else if ("--help".equalsIgnoreCase(argv[i])) {
                Usage();
            } else {
                System.err.println("\nUnknown parameter: " + argv[i] + "\n");
                Usage();
            }
        }

        if (port == -1) {
            port = 6379;
        }

        System.out.println("");
        System.out.println("EndPoint: " + host + ":" + port);
        System.out.println("Concurrent clients: " + threads);
        System.out.println("Test mode: " + (branch == 1 ? "performance" : (branch == 2 ? "stable" : "unknown")));
        System.out.println("");

        try {
            switch (branch) {
                case 1:
                    switch (cmd) {
                        case 0:
                            pString(host, port, threads, password);
                            break;
                        case 1:
                            pList(host, port, threads, password);
                            break;
                        case 2:
                            pSet(host, port, threads, password);
                            break;
                        case 3:
                            pHash(host, port, threads, password);
                            break;
                        case 4:
                            pZset(host, port, threads, password);
                            break;
                        case 5:
                            pPublish(host, port, threads, password);
                            break;
                        case 10:
                            pCustom(host, port, threads, password, custom_cmds);
                            break;
                        case 100:
                            pString(host, port, threads, password);
                            pList(host, port, threads, password);
                            pSet(host, port, threads, password);
                            pHash(host, port, threads, password);
                            pZset(host, port, threads, password);
                            pPublish(host, port, threads, password);
                            break;
                        default:
                            System.out.println("Unknown command");
                            Usage();
                    }
                    break;
                case 2:
                    stable(host, port, password);
                    break;
                default:
                    Usage();
                    break;
            }
        } catch (Throwable t) {
            System.out.println("Error occur: " + t);
            Usage();
        }
    }
}
